/*
 * =====================================================================================
 *
 *       Filename:  cl_lockfree_queue.hxx
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  02/17/2011 12:58:38 AM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Ken Chen (Ken), chen_kenz@emc.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#ifndef _CL_LOCKFREE_QUEUE_HXX_
#define _CL_LOCKFREE_QUEUE_HXX_

#include <atomic>

namespace neo
{
#define CL_CACHE_LINE_SIZE 64

template<typename T>
class CLLockFreeQueue
{
private:
  struct Node
  {
    Node(T* val)
      : m_value(val), 
      m_next(NULL)
    {
    }

    T* m_value;
    std::atomic<Node*> m_next;
    // cache padding
    char pad[CL_CACHE_LINE_SIZE - (sizeof(T*)+sizeof(std::atomic<Node*>))%CL_CACHE_LINE_SIZE];
  };

  char pad0[CL_CACHE_LINE_SIZE];

  Node *m_first;
  char pad1[CL_CACHE_LINE_SIZE - sizeof(Node*)];

  std::atomic<bool> m_consumerLock;
  char pad2[CL_CACHE_LINE_SIZE - sizeof(std::atomic<bool>)];

  Node *m_last;
  char pad3[CL_CACHE_LINE_SIZE - sizeof(Node*)];

  std::atomic<bool> m_producerLock;
  char pad4[CL_CACHE_LINE_SIZE - sizeof(std::atomic<bool>)];

public:
  CLLockFreeQueue()
  {
    m_first = m_last = new Node(NULL); 
    m_producerLock = m_consumerLock = false;
  }

  ~CLLockFreeQueue()
  {
    while(m_first != NULL)
    {
      Node* tmp = m_first;
      m_first = tmp->m_next;
      delete tmp->m_value;
      delete tmp;
    }
  }

  void push(const T& t)
  {
    Node* tmp = new Node(new T(t));
    while(m_producerLock.exchange(true))
    {
    }
    m_last->m_next = tmp;
    m_last = tmp;
    m_producerLock = false;
  }

  void push(T&& t)
  {
    Node* tmp = new Node(new T(std::move(t)));
    while(m_producerLock.exchange(true))
    {
    }
    m_last->m_next = tmp;
    m_last = tmp;
    m_producerLock = false;
  }

  bool pop(T& t)
  {
    while(m_consumerLock.exchange(true))
    {
    }
    Node* theFirst = m_first;
    Node* theNext = theFirst->m_next;
    if(theNext != NULL)
    {
      T* val = theNext->m_value;
      theNext->m_value = NULL;
      m_first = theNext;
      m_consumerLock = false;
      t = std::move(*val);
      delete val;
      delete theFirst;
      return true;
    }
    m_consumerLock = false;
    return false;
  }
};
} // neo

#endif //_CL_LOCKFREE_QUEUE_HXX_

